package com.burges.net.tableAPIAndSQL.time

import java.util

import com.sun.xml.internal.ws.policy.privateutil.PolicyUtils.Collections
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.{datastream, environment}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.{TableEnvironment, TableSchema, Types}
import org.apache.flink.table.sources.tsextractors.ExistingField
import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps
import org.apache.flink.table.sources.{DefinedRowtimeAttributes, RowtimeAttributeDescriptor, StreamTableSource}
import org.apache.flink.types.Row

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/14 
  * 描述     时间概念EventTime指定
  */
object EventTimeDefineDemo {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		val tableSourceEnv = TableEnvironment.getTableEnvironment(environment)
		// 获取输入数据集
		val inputStream: DataStream[String] = environment.fromElements("flink", "hadoop")
		// 调用DataStreamAPI的assignTimestampsAndWatermarks指定EventTIme和watermark信息
		val stream: DataStream[(String,String)] = inputStream.assignTimestampsAndWatermarks(...)

		/**
		  * 在DataStream转换Table的过程中定义
		  */
		//在TableSchema末尾使用'event_time.rowtime定义EventTime字段'
		//系统会从TableEnvironment中获取EventTime信息
		tableSourceEnv.fromDataStream(stream, 'id, 'var1, 'event_time.rowtime)

		//调用DataStreamAPI的assignTimestampsAndWatermarks指定EventTime和watermark信息，并在DataStream中将第一个字段提取出来并指定为EventTime字段
		val watermarkStream: DataStream[(Long, String, String)] = inputStream.assignTimestampsAndWatermarks(...)
		val table = tableSourceEnv.fromDataStream(stream, 'event_time.rowtime, 'id, 'var1)


		//通过TableSource函数定义使用方式

		//注册输入数据源
		tableSourceEnv.registerTableSource("InputEvent", new InputEventSource)
		//在窗口中使用输入数据源，并基于TableSource中定义的EventTime字段创建窗口
		tableSourceEnv.scan("InputEvent").window("Tumble over 10 mins on 'event_time as window'")

	}

	//通过TableSource函数定义
	//定义InputEventSource创建外部数据源
	//并实现DefinedRowtimeAttributes接口来定义EventTime时间属性
	class InputEventSource extends StreamTableSource[Row] with DefinedRowtimeAttributes{

		// 实现StreamTableSource接口中的getDataStream方法定义输入数据源
		override def getDataStream(execEnv: environment.StreamExecutionEnvironment): datastream.DataStream[Row] = {
			//定义获取DataStream数据集的逻辑
			val inputStream: DataStream[(String, String, Long)] = ...
			// ....
			//指定数据集中的EventTime时间信息和Watermark
			val stream = inputStream.assignTimestampsAndWatermarks(...)
			stream
		}

		// 定义TableAPI中的时间属性信息
		override def getRowtimeAttributeDescriptors: util.List[RowtimeAttributeDescriptor] = {
			//创建基于event_time的RowtimeAttributeDescriptor，确定时间属性信息
			val rowtimeAttrDesc = new RowtimeAttributeDescriptor(
				"event_time", //时间属性名称
				new ExistingField("event_time"),
				new AscendingTimestamps
			)
			val rowTimeAttrDescList = Collections.singletonList(rowtimeAttrDesc)
			rowTimeAttrDescList
		}

		// 定义数据集字段名称和类型
		override def getReturnType: TypeInformation[Row] = {
			val names: Array[String] = Array[String]("id", "value", "event_time")
			val types: Array[TypeInformation[_]] = Array[TypeInformation[_]](Types.STRING, Types.STRING, Types.LONG)
			Types.ROW(names, types)
		}

		override def getTableSchema: TableSchema = ???
	}

}
